package com.ultrapower.request;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import com.sematext.hbase.wd.AbstractRowKeyDistributor;
import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.bean.AcceptApplicationBean;
import com.ultrapower.bean.ApplicationBean;
import com.ultrapower.buffer.AutomaticBuffer;
import com.ultrapower.buffer.Buffer;
import com.ultrapower.buffer.OffsetFixedBuffer;
import com.ultrapower.util.CommonUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.TimeSlot;
import com.ultrapower.util.rowkey.RangeOneByteSimpleHash;
/**
 * 
 * @author wangy
 *
 */
public class HostApplicationMap_Ver2Table extends BaseRequest {
	private int scanCacheSize = 10;
	private TimeSlot timeSlot = new TimeSlot();
	private AbstractRowKeyDistributor acceptApplicationRowKeyDistributor = new RowKeyDistributorByHashPrefix(
			new RangeOneByteSimpleHash(0, 24, 4));
	
	public static void main(String[] args) {
		BaseRequest table=new HostApplicationMap_Ver2Table();
		table.execute();
	}

	@Override
	void execute() {
		TableName hostApplicationMapTableName = this.getTableName(HBaseTables.HOST_APPLICATION_MAP_VER2_STR);
		this.getHbaseTemplate().find(hostApplicationMapTableName, createScan(),acceptApplicationRowKeyDistributor, (ResultScanner scanner) -> {

			scanner.forEach(r -> {
				  	final List<AcceptApplicationBean> acceptApplicationList = new ArrayList<>(r.size());
				  	Buffer row =new  OffsetFixedBuffer(r.getRow());
				  	String appName = row.readPadStringAndRightTrim(HBaseTables.APPLICATION_NAME_MAX_LEN);
				  	System.out.println(appName);
			        for (Cell cell : r.rawCells()) {
			            AcceptApplicationBean acceptedApplication = createAcceptedApplication(cell);
			            acceptApplicationList.add(acceptedApplication);
			            System.out.println("\t"+acceptedApplication);
			        }
			});
			return null;
		});
	}
	
	private AcceptApplicationBean createAcceptedApplication(Cell cell) {
        Buffer reader = new OffsetFixedBuffer(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
        String host = reader.readPrefixedString();
        String bindApplicationName = reader.readPrefixedString();
        short bindServiceTypeCode = reader.readShort();
        final ApplicationBean bindApplication = new ApplicationBean(bindApplicationName, bindServiceTypeCode);
        return new AcceptApplicationBean(host, bindApplication);
    }

	@Override
	protected Scan createScan() {
		from=1532681282546L;
		to=  1532683128546L;
		final long startTime = CommonUtils.reverseTimeMillis(timeSlot.getTimeSlot(from));
		final long endTime = CommonUtils.reverseTimeMillis(timeSlot.getTimeSlot(to) + 1);
		//
		final byte[] startKey = createKey(applicationName, srviceTypeCode, endTime);
		final byte[] endKey = createKey(applicationName, srviceTypeCode, startTime);

		Scan scan = new Scan();
		scan.setCaching(this.scanCacheSize);
		scan.setStartRow(startKey);
		scan.setStopRow(endKey);
		scan.setId("HostApplicationScan_Ver2");

		return scan;
	}

	private byte[] createKey(String appName, short serviceTypeCode, long time) {
		Buffer buffer = new AutomaticBuffer();
		buffer.putPadString(appName, HBaseTables.APPLICATION_NAME_MAX_LEN);
		buffer.putShort(serviceTypeCode);
		buffer.putLong(time);
		return buffer.getBuffer();
	}

}
